package com.knight.hadoop.day04.mq.queue;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ProductTool {
	private String user = ActiveMQConnection.DEFAULT_USER;
	private String password = ActiveMQConnection.DEFAULT_PASSWORD;
	private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	private String subject = "mytopic";
	private Destination destination = null;
	private Connection connection = null;
	private Session session = null;
	private MessageProducer messageProducer = null;

	/**
	 * 初始化
	 * 
	 * @throws Exception
	 */
	private void init() throws Exception {
		ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(user, password, url);
		connection = activeMQConnectionFactory.createConnection();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		destination = session.createTopic(subject);
		messageProducer = session.createProducer(destination);
		messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	}

	/**
	 * 发送消息
	 * 
	 * @param message
	 * @throws Exception
	 */
	public void produceMessage(String message) throws Exception {
		init();
		TextMessage textMessage = session.createTextMessage(message);
		connection.start();
		System.out.println("Producer:->Sending message: " + message);
		messageProducer.send(textMessage);
		System.out.println("Producer:->Message sent complete!");
	}

	/**
	 * 关闭连接
	 * 
	 * @throws JMSException
	 */
	public void close() throws JMSException {
		System.out.println("Producer:->Closing connection");
		if (messageProducer != null) {
			messageProducer.close();
		}
		if (session != null) {
			session.close();
		}
		if (connection != null) {
			connection.close();
		}
	}

}
